package com.pxene.dmp.mapper;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.google.gson.Gson;
import com.pxene.dmp.common.URLUtils;

public class GetTicketInfoMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
	private Log log = LogFactory.getLog(GetTicketInfoMapper.class);
	private  URL urlParser;
	private LongWritable resultKey = new LongWritable();
	private Text resultVal = new Text();
	private  Map<String, String[]> apis = new HashMap<String, String[]>();

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
		// 设置灵活接口配置文件名
		for (String api : conf.getStrings("apis")) {
			String[] array = api.split("\\|");
			if (array.length > 2) {
				String code = array[0];
				String role = array[1];
				String[] regexs = new String[array.length-2];
				for (int i=2; i<array.length; i++) {
					String param = "${" + array[i] + "}";
					if (role.contains(param)) {
						regexs[i-2] = role.substring(role.indexOf("/"), role.length()).replace(param, "(.*)").replaceAll("\\$\\{.*?\\}", ".*");
					} else {
						regexs[i-2] = array[i];
					}
				}
				apis.put(code, regexs);
			}
		}
	}

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String[] columns = value.toString().split("\\|");
		if (columns.length != 7) {
			return;
		}
		String imsi = columns[0];
		String province = columns[1];
		String starttime = columns[3];
		String domain = columns[4];
		String url = columns[5];
		String dataType = columns[6];
		
		try {
			urlParser = new URL(url);
		} catch(MalformedURLException e) {
			return;
		}
		
		Map<String, String> map = URLUtils.getQueryMap(urlParser.getQuery());
		
		String[] regexs = apis.get(domain);
		List<String> params =new  ArrayList<String>();
		for (int i=0; i<regexs.length; i++) {
			if (regexs[i].contains("(.*)")) {
				 params.add(URLUtils.regexpExtract(url, regexs[i]));
			} else {
				if(!regexs[i].contains("^")){
					 params.add(map.get(regexs[i]) == null ? "" : map.get(regexs[i]).trim());
				}else{ //如果请求值为json串，按照json串的方式解析
					String[] tempParams=regexs[i].split("^");
					String jsonStr = map.get(tempParams[0]) == null ? "" : map.get(tempParams[0]).trim();
					for(int j=1;j<tempParams.length;j++){
						Map<String, String> jsonmap = jsonToMap(jsonStr);
						params.add( (String) (jsonmap.get(tempParams[j])==null ? "" : jsonmap.get(tempParams[j])==null)    );
					}
				}
			}
		}
		
		String paramStr = StringUtils.join(params, ";");
		if (paramStr.length() > params.size()) {
			resultKey.set(new Date().getTime());
			resultVal.set(imsi + "|" + starttime + "|" + domain 
					+ "|" + paramStr + "|" + province + "|" + dataType);
			context.write(resultKey, resultVal);
			log.info("TTTT---"+paramStr);
		}
		
	}
	
	
	
	
	 /**
     * 将传递近来的json对象转换为Map集合
     */
	 private static Map<String, String> jsonToMap(String jsonStr) {
	        Map<String, String> ObjectMap = null;
	        Gson gson = new Gson();
	        java.lang.reflect.Type type = new com.google.gson.reflect.TypeToken<Map<String, String>>() {}.getType();
	        ObjectMap = gson.fromJson(jsonStr, type);
	        return ObjectMap;
	  }
	
}
